package com.example.testkafka.producer.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/producer")
public class ProducerController {

    private static final Logger logger = LoggerFactory.getLogger(ProducerController.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 批量同步发送
     * @param msg
     * @return
     */
    @RequestMapping("/sendSync")
    @ResponseBody
    public Map<String,Object> sendSync(String msg) {
        Map<String,Object> appResult = new HashMap<>();
        for(int i = 0;i < 10;i ++){
            ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("test_topic", i + "",msg + ":" + i);
//            listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
//                @Override
//                public void onFailure(Throwable throwable) {
//                    System.out.println(Thread.currentThread().getName() + ":发送消息报错" + throwable);
//                }
//
//                @Override
//                public void onSuccess(SendResult<String, String> stringStringSendResult) {
//                    System.out.println(Thread.currentThread().getName() + ":发送消息成功");
//                }
//            });
            try {
                SendResult<String, String> sendResult = listenableFuture.get();
                RecordMetadata recordMetadata = sendResult.getRecordMetadata();
                System.out.println(recordMetadata.toString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.out.println(Thread.currentThread().getName() + ":执行完毕");
        appResult.put("status","SUCCESS");
        return appResult;
    }

    /**
     * 异步发送
     * @param msg
     * @return
     */
    @RequestMapping("/sendAsync")
    @ResponseBody
    public Map<String,Object> sendAsync(String msg) {

        Map<String,Object> appResult = new HashMap<>();
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("test_topic", msg);
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                RecordMetadata recordMetadata = stringStringSendResult.getRecordMetadata();
                System.out.println("********************topic:" + recordMetadata.topic() + ",partition:" + recordMetadata.partition() + ",offset:" + recordMetadata.offset());
            }
        });
        appResult.put("status","SUCCESS");
        return appResult;
    }

    /**
     * 事务方式1
     * @param msg
     * @param occurException
     * @return
     */
    @RequestMapping("/testTransaction")
    @ResponseBody
    public Map<String,Object> testTransaction(String msg,boolean occurException) {

        Map<String,Object> appResult = new HashMap<>();

        String result = kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, String, String>() {
            @Override
            public String doInOperations(KafkaOperations<String, String> operations) {

                for(int i = 0;i < 10;i ++){
                    //线程睡一会儿,不然执行太快,未填满缓冲区就报错,消费者收不到(read_uncommitted)
                    try {
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if(i == 8 && occurException){
                        System.out.println(1 / 0);
                    }
                    ListenableFuture<SendResult<String, String>> future = operations.send("test_topic", i + "",msg + ":" + i);
                    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                        @Override
                        public void onFailure(Throwable throwable) {
                            System.out.println("发送消息报错" + throwable);
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> stringStringSendResult) {
                            System.out.println("发送消息成功");
                        }
                    });
                }
                return "hello";
            }
        });
        System.out.println("===================" + result);
        appResult.put("status","SUCCESS");
        return appResult;
    }
}
